package com.myrice.jms.impl;

import java.nio.ByteBuffer;

import com.myrice.core.Connection;
import com.myrice.core.MessageOutput;
import com.myrice.core.Session;
import com.myrice.filter.FilterAdapter;
import com.myrice.filter.IFilterChain.FilterChain;
import com.myrice.filter.IProtocolDecodeFilter;
import com.myrice.filter.IProtocolEncodeFilter;
import com.myrice.jms.IMsgProtocolFilter;
import com.myrice.jms.MessageFactory;
import com.myrice.jms.MsgRequest;
import com.myrice.util.ByteBufferUtils;

public class DefaultMsgProtocolFilter extends FilterAdapter implements
		IMsgProtocolFilter {
	private MessageFactory messageFactory;

	public void setMessageFactory(MessageFactory messageFactory) {
		this.messageFactory = messageFactory;
	}

	public void messageEncode(Connection conn, Object message,
			MessageOutput out, FilterChain<IProtocolEncodeFilter> chain)
			throws Exception {
		Session session = conn.getSession();
		IProtocolEncodeFilter protocol = (IProtocolEncodeFilter) session
				.getAttribute(Session.IO_PROTOCOL_ENCODE);
		if (protocol != null) {
			protocol.messageEncode(conn, message, out, chain);
			return;
		}
		if (message instanceof ByteBuffer) {
			ByteBuffer msg = (ByteBuffer) message;
			// 是否有8个字节剩余空间
			if (msg.capacity() - msg.limit() >= 8) {
				ByteBufferUtils.offset(msg, 4);
				msg.limit(msg.limit() + 8);
				int length = msg.remaining();
				msg.putInt(msg.position(), length);// 头部
				msg.putInt(msg.position() + length - 4, length);// 尾部
				message = msg;
			} else {
				int length = 8 + msg.remaining();
				ByteBuffer packet = ByteBuffer.allocate(length);
				packet.putInt(length).put(msg).putInt(length).flip();// 数据包长+数据包内容+校验长度
				message = packet;
			}
		}

		if (chain.hasNext()) {
			chain.nextFilter().messageEncode(conn, message, out,
					chain.getNext());
		} else {
			out.putLast(message);
		}
	}

	public boolean messageDecode(Connection conn, ByteBuffer in,
			MessageOutput out, FilterChain<IProtocolDecodeFilter> chain) {
		Session session = conn.getSession();
		if (session instanceof DefaultMsgSession) {
			((DefaultMsgSession) session).updateLastTime();
		}
		IProtocolDecodeFilter protocol = (IProtocolDecodeFilter) session
				.getAttribute(Session.IO_PROTOCOL_DECODE);
		if (protocol != null) {
			return protocol.messageDecode(conn, in, out, chain);
		}
		boolean suc = false;
		for (; in.position() >= 4 && in.getInt(0) <= in.position()
				&& in.getInt(0) == in.getInt(in.getInt(0) - 4);) {
			suc = true;
			in.flip();
			int length = in.getInt();// 读消息长度(包括此字段)
			ByteBuffer msg = ByteBuffer.allocate(length - 8);// 数据
			in.get(msg.array());// 读内容
			in.getInt();// 校验长度
			in.compact();/* 压缩缓冲区 */
			if (chain.hasNext() == false
					|| !chain.nextFilter().messageDecode(conn, msg, out,
							chain.getNext())) {
				out.putLast(createRequest((DefaultMsgSession) session, msg));
				if (out.isFulled())
					return true;
			}
		}
		if (suc == false) {
			if (chain.hasNext()) {
				return chain.nextFilter().messageDecode(conn, in, out,
						chain.getNext());
			}
		}
		return suc;
	}

	private MsgRequest createRequest(DefaultMsgSession session, ByteBuffer msg) {
		return messageFactory.create(session, msg);
	}

	public MessageFactory getMessageFactory() {
		return messageFactory;
	}
}
